package com.ln.reactive.msg.test;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class DefaultSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;

    private int count = 0;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
    }

    @Override
    public void onNext(T t) {
        // 当到达阈值的之后，取消Publisher 给当前的 Subscriber发送数据
        if (count++ > 2) {
            subscription.cancel();
            return;
        }
        System.out.println("收到的消息" + t);
    }

    @Override
    public void onError(Throwable t) {
        System.err.println("发生了异常" + t);
    }

    @Override
    public void onComplete() {
        System.out.println("收到数据完成");
    }
}
